%%%-----------------------------------------------------------------------------
%%% @doc
%%% emqttd vm monitor.
%%%
%%% @end
%%%-----------------------------------------------------------------------------
-module(monitor).

-behavior(gen_server).

-export([start_link/1, dispatch/2]).

-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-define(RECONNECT_COUNT, 20).

-record(state, {ok, cluster, strategy, reconnect_count}).

%%------------------------------------------------------------------------------
%% @doc
%% Start emqttd monitor.
%%
%% @end
%%------------------------------------------------------------------------------
start_link({Cluster, Strategy}) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [{Cluster, Strategy}], []).

%%%=============================================================================
%%% gen_server callbacks
%%%=============================================================================

init([{Cluster, Strategy}]) ->
    process_flag(trap_exit, true),
    %% 订阅mnesia事件
    {ok, _} = mnesia:subscribe(system),
    erlang:system_monitor(self(), [{long_gc, 5000}, {large_heap, 1000000}, busy_port]),
    {ok, #state{cluster = Cluster, strategy = Strategy, reconnect_count = ?RECONNECT_COUNT}}.

handle_call(Request, _From, State) ->
    lager:error("unexpected request: ~p", [Request]),
    {stop, {error, unexpected_request}, State}.

handle_cast(Msg, State) ->
    lager:error("unexpected msg: ~p", [Msg]),
    {noreply, State}.

handle_info({mnesia_system_event, {inconsistent_database, running_partitioned_network, _}}, #state{cluster = Cluster, strategy = Strategy} = State) ->
    restart_node_mnesia(Cluster, Strategy),
    {noreply, State};

handle_info({mnesia_system_event, {mnesia_down, Node}}, #state{cluster = Cluster, strategy = Strategy} = State) ->
    lager:error("---~p--~p-~p~n", [node(), Node, mnesia:system_info(running_db_nodes)]),
    case Node =/= node() of
        false ->
            {Type, _, _} = lists:keyfind(node(), 2, Cluster),
            dispatch(Cluster, Type),
            {noreply, State};    
        true ->
            {noreply, State}
    end;


handle_info({monitor, GcPid, long_gc, Info}, State) ->
    lager:error("long_gc: gcpid = ~p, ~p ~n ~p", [GcPid, process_info(GcPid, 
		[registered_name, memory, message_queue_len,heap_size,total_heap_size]), Info]),
    {noreply, State};

handle_info({monitor, GcPid, large_heap, Info}, State) ->
    lager:error("large_heap: gcpid = ~p,~p ~n ~p", [GcPid, process_info(GcPid, 
		[registered_name, memory, message_queue_len,heap_size,total_heap_size]), Info]),
    {noreply, State};

handle_info({monitor, SusPid, busy_port, Port}, State) ->
    lager:error("busy_port: suspid = ~p, port = ~p", [process_info(SusPid, 
		[registered_name, memory, message_queue_len,heap_size,total_heap_size]), Port]),
    {noreply, State};

handle_info(Info, State) ->
    lager:error("unexpected info: ~p", [Info]),
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.


dispatch(Cluster, Type) ->
    mnesia:stop(),
    mnesia:start(),
    case Type of
        remote ->
            schema_db:remote_dispatch(Cluster);
        disc   ->
            schema_db:disc_dispatch(Cluster);
        ram    ->
            schema_db:ram_dispatch(Cluster);
        _ ->
            schema_db:mnesia_init(Cluster)
    end.
  

restart_node_mnesia(Cluster, Strategy) ->
    Master = lists:keyfind(disc, 1, Cluster),
    {Type, MasterNode, _} = Master,
    case node() =:= MasterNode of
        true ->
            lists:foreach(
                fun({_, Node, _}) ->
                    rpc:cast(Node, ?MODULE, dispatch, [Cluster, Type])
                end, lists:delete(Master, Cluster));
        false ->
            ok
    end.

